package com.abyss.transformation;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.operators.DataSource;
import org.apache.flink.api.java.operators.MapOperator;

import java.text.SimpleDateFormat;

/**
 * @author Abyss
 * @date 2020/10/4
 * @description 使用map操作，读取apache.log文件中的字符串数据转换成ApacheLogEvent对象
 */
public class MapDemo {
    public static void main(String[] args) throws Exception {
        // 1. Env
        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();

        // 2. Source 读取apache.log文件内容
        DataSource<String> fileSource = env.readTextFile("/Users/abyss/Dev/toys/flink/H-flink-learn/src/main/resources/apache.log");

        // 3. 处理数据将字符串转变为ApacheEvent对象
        MapOperator<String, ApacheEvent> apacheEvents = fileSource.map(new MapFunction<String, ApacheEvent>() {
            SimpleDateFormat simpleDateFormat = new SimpleDateFormat("dd/MM/yyyy:HH:mm:ss");

            @Override
            public ApacheEvent map(String value) throws Exception {
                String[] strings = value.split(" ");
                String ip = strings[0];
                int userId = Integer.parseInt(strings[1]);
                long ts = simpleDateFormat.parse(strings[2]).getTime();
                String method = strings[3];
                String url = strings[4];

                return new ApacheEvent(ip, userId, ts, method, url);
            }
        });

        // 4. 输出
        apacheEvents.print();
    }

    /*
    定义POJO类存储apache.log数据的schema
     */
    public static class ApacheEvent {
        private String ip;
        private Integer userId;
        private Long ts;
        private String method;
        private String url;

        public ApacheEvent(String ip, Integer userId, Long ts, String method, String url) {
            this.ip = ip;
            this.userId = userId;
            this.ts = ts;
            this.method = method;
            this.url = url;
        }

        public ApacheEvent() {}

        @Override
        public String toString() {
            return "ApacheEvent{" +
                    "ip='" + ip + '\'' +
                    ", userId=" + userId +
                    ", ts=" + ts +
                    ", method='" + method + '\'' +
                    ", url='" + url + '\'' +
                    '}';
        }

        public String getIp() {
            return ip;
        }

        public void setIp(String ip) {
            this.ip = ip;
        }

        public Integer getUserId() {
            return userId;
        }

        public void setUserId(Integer userId) {
            this.userId = userId;
        }

        public Long getTs() {
            return ts;
        }

        public void setTs(Long ts) {
            this.ts = ts;
        }

        public String getMethod() {
            return method;
        }

        public void setMethod(String method) {
            this.method = method;
        }

        public String getUrl() {
            return url;
        }

        public void setUrl(String url) {
            this.url = url;
        }
    }
}
